package edu.neu.swc.skynet.service.impl;

import com.alibaba.fastjson.JSON;
import edu.neu.swc.skynet.common.SearchParams;
import edu.neu.swc.skynet.common.TableId;
import edu.neu.swc.skynet.dao.SkyNetMovieDao;
import edu.neu.swc.skynet.document.MovieDoc;
import edu.neu.swc.skynet.entity.SkyNetMovie;
import edu.neu.swc.skynet.movie.MovieInfo;
import edu.neu.swc.skynet.movie.MovieShow;
import edu.neu.swc.skynet.service.IMovieService;
import edu.neu.swc.skynet.vo.PageMovieShow;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.IterableUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

@Service
@Slf4j
@Transactional(rollbackFor = Exception.class)
public class MovieServiceImpl implements IMovieService {

    private final SkyNetMovieDao movieDao;
    private final RestHighLevelClient restHighLevelClient;

    public MovieServiceImpl(SkyNetMovieDao movieDao,
                            RestHighLevelClient restHighLevelClient) {
        this.movieDao = movieDao;
        this.restHighLevelClient = restHighLevelClient;
    }

    @Override
    public List<MovieInfo> getMovieInfoByTableId(TableId tableId) {

        //1. 解析 Id 列表
        //这里不能从缓存中取数据，需要访问数据库
        List<Long> ids = tableId.getIds().stream()
                .map(TableId.Id::getId)
                .collect(Collectors.toList());

        log.info("get movies by ids:[{}]", JSON.toJSONString(
                ids
        ));

        //2. 查询列表
        List<SkyNetMovie> movieInfos = IterableUtils.toList(
                movieDao.findAllById(ids)
        );

        //3. 列表转化，使用 函数式接口 转化
        return movieInfos.stream()
                .map(SkyNetMovie::toMovieInfo)
                .collect(Collectors.toList());
    }

    @Override
    public List<MovieShow> getMovieShowByTableId(TableId tableId) {

        //1. 解析 Id 列表
        //TODO 未来这里添加优先访问 ES 的业务逻辑
        List<Long> ids =tableId.getIds().stream()
                .map(TableId.Id::getId)
                .collect(Collectors.toList());

        log.info("get movies by ids:[{}]", JSON.toJSONString(
                ids
        ));

        //2. 查询列表
        List<SkyNetMovie> movieInfos = IterableUtils.toList(
                movieDao.findAllById(ids)
        );

        //3. 列表转化，使用 函数式接口 转化
        return movieInfos.stream()
                .map(SkyNetMovie::toMovieShow)
                .collect(Collectors.toList());
    }

    @Override
    public PageMovieShow getMovieShowByPage(int page) {

        //分页不能从Redis Cache中取得
        if(page <= 1){
            page = 1; // 默认是第一页
        }

        // 这里分页的规则（可以自由修改）：一页包含10条数据，按照商品ID倒序排列
        Pageable pageable = PageRequest.of(
                page - 1,
                10,
                Sort.by("id").descending()
        );

        Page<SkyNetMovie> orderPage = movieDao.findAll(pageable);

        // 是否还有更多页: 总页数 是否大于 当前打印页
        boolean hasMore = orderPage.getTotalPages() > page;

        // 列表转换
        return new PageMovieShow(
                orderPage.getContent().stream()
                        .map(SkyNetMovie::toMovieShow)
                        .collect(Collectors.toList()),
                hasMore
        );
    }

    /**
     * <h2>查询字段实现</h2>
     * 1. 构造查询请求
     * 2. 解析响应参数
     * 3. 构造返回结果
     * @param searchParams ES 专用查询对象
     * @return 分页查询信息
     */
    @Override
    public PageMovieShow searchMovieShowByPage(SearchParams searchParams) {

        //等待接收的结果
        List<MovieShow> movieShows;
        boolean hasMore = false;
        try {
            //1. 构造查询请求
            BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();

            //1.1 关键字搜索，match查询
            String key = searchParams.getKey();

            if(StringUtils.isNotBlank(key)){
                // 关键字非空，按照关键字查询
                // DSL:
                boolQuery.must(QueryBuilders.matchQuery("all", key));
            }else{
                // 为空，查询所有
                boolQuery.must(QueryBuilders.matchAllQuery());
            }

            // 准备查询请求
            SearchRequest request = new SearchRequest("sky_net_movie");
            request.source().query(boolQuery);

            // 准备分页信息
            int page = searchParams.getPage();
            int size = searchParams.getSize();
            request.source().from((page - 1) * size).size(size);

            // 发送请求，接收数据
            SearchResponse response = restHighLevelClient.search(request,
                    RequestOptions.DEFAULT);

            //2. 解析响应
            SearchHits searchHits = response.getHits();
            long total = searchHits.getTotalHits().value;
            hasMore = ((long) page * size) < total;
            SearchHit[] hits = searchHits.getHits();
            List<MovieDoc> movieDocs = new ArrayList<>();

            for (SearchHit hit : hits) {
                String movieDocJson = hit.getSourceAsString();
                MovieDoc movieDoc = JSON.parseObject(
                        movieDocJson,
                        MovieDoc.class
                );
                movieDocs.add(movieDoc);
            }

            movieShows = movieDocs.stream()
                    .map(MovieDoc::toMovieShow)
                    .collect(Collectors.toList());

        } catch (IOException e) {
            e.printStackTrace();
            movieShows = Collections.emptyList();
        }
        //3. 解析Page对象
        return new PageMovieShow(
                movieShows,
                hasMore
        );
    }
}
